package spring.cloud.tasks.tasks_assign_manager.task_item_assign;

import com.alibaba.fastjson.JSON;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import spring.cloud.tasks.common.BatchTaskInstanceInfo;
import spring.cloud.tasks.common.BatchTaskReStartInfo;
import spring.cloud.tasks.common.cron_expression.DateUtils;
import spring.cloud.tasks.common.node.BlockNodePaths;
import spring.cloud.tasks.common.node.ExecutorNodePath;
import spring.cloud.tasks.common.node.TaskNodePath;
import spring.cloud.tasks.common.task_item_allot.OnlineExecutor;
import spring.cloud.tasks.common.task_item_allot.TaskItem;
import spring.cloud.tasks.tasks_assign_manager.domain.WaitedAssignTaskInfo;
import spring.cloud.tasks.tasks_assign_manager.schedule.CronExpressionContext;
import spring.cloud.tasks.tasks_assign_manager.schedule.CronExpressionContextManager;
import spring.cloud.tasks.tasks_assign_manager.schedule.TaskPreAssignTimePoint;
import spring.cloud.tasks.tasks_assign_manager.util.DateTimeUtil;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Setter
@Getter
public class TaskItemAssignUtils {
    //##<!-我是优美的功能模块分割线->##

    /**
     * 获取该 nameSpace下的所有task id list
     */
    public static List<String> getTaskIdList(CuratorFramework curatorFramework) throws Exception {
        if (curatorFramework.checkExists().forPath(TaskNodePath.tasks) == null) {
            curatorFramework.create().creatingParentsIfNeeded().forPath(TaskNodePath.tasks);
        }
        List<String> taskIdList = new ArrayList<>();
        List<String> $taskIdList = curatorFramework.getChildren().forPath(TaskNodePath.tasks);
        if ($taskIdList != null) {
            taskIdList.addAll($taskIdList);
        }
        return taskIdList;
    }

    /**
     * 获取该 nameSpace下的所有enable task id list
     */
    public static List<String> getEnableTaskIdList(CuratorFramework curatorFramework, List<String> taskIdList) throws Exception {
        List<String> enableTaskIdList = new ArrayList<>();
        for (int i = 0; i < taskIdList.size(); i++) {
            String taskId = taskIdList.get(i);
            if (curatorFramework.checkExists().forPath(BlockNodePaths.getEnableTaskNodePath(taskId)) != null) {
                enableTaskIdList.add(taskId);
            }
        }
        return enableTaskIdList;
    }
    //##<!-我是优美的功能模块分割线->##

    /**
     * 域获取最新的执行器。该执行器不包括支持的任务和已经分配的任务项
     */
    public static List<OnlineExecutor> getLastOnlineExecutorList(CuratorFramework curatorFramework) throws Exception {
        if (!(curatorFramework.checkExists().forPath(ExecutorNodePath.getExecutorsNodePath()) != null)) {
            return new ArrayList<>();
        }
        //获取所有正在运行的Executor
        List<String> executorIdList = curatorFramework.getChildren().forPath(ExecutorNodePath.getExecutorsNodePath());
        if (executorIdList == null) {
            return new ArrayList<>();
        }
        List<OnlineExecutor> onlineExecutorList = new ArrayList<>();
        for (int i = 0; i < executorIdList.size(); i++) {
            String executorId = executorIdList.get(i);
            String ip = TaskItemAssignUtils.getIp(curatorFramework, executorId);
            if (ip != null) {
                OnlineExecutor onlineExecutor = new OnlineExecutor();
                onlineExecutor.setExecutorId(executorId);
                onlineExecutor.setIp(ip);
                onlineExecutor.setOwnTraffic(TaskItemAssignUtils.executorIsOwnTraffic(curatorFramework, executorId));
                List<String> taskIdList = new ArrayList<>();
                if (curatorFramework.checkExists().forPath(BlockNodePaths.getExecutorTasksNodePath(executorId)) != null) {
                    taskIdList = curatorFramework.getChildren().forPath(BlockNodePaths.getExecutorTasksNodePath(executorId));
                }
                onlineExecutor.setSupportedTaskIdList(taskIdList);
                onlineExecutor.setAssignedTaskItemList(new ArrayList<>());//init
                onlineExecutorList.add(onlineExecutor);
            }
        }
        return onlineExecutorList;
    }

    /**
     * @return 可能返回为空
     * @throws Exception
     */
    public static String getIp(CuratorFramework curatorFramework, String executorId) throws Exception {
        String ip = null;
        String executorIpNodePath = ExecutorNodePath.getExecutorIpNodePath(executorId);
        if (curatorFramework.checkExists().forPath(ExecutorNodePath.getExecutorIpNodePath(executorId)) != null) {
            byte[] bytes = curatorFramework.getData().forPath(executorIpNodePath);
            if (bytes != null) {
                ip = new String(bytes, StandardCharsets.UTF_8);
            }
        }
        return ip;
    }

    public static boolean executorIsOwnTraffic(CuratorFramework curatorFramework, String executorId) throws Exception {
        byte[] bytes = curatorFramework.getData().forPath(ExecutorNodePath.getExecutorOwnTrafficNodePath(executorId));//Y有流量 N没有流量  默认是Y
        return bytes != null && Boolean.parseBoolean(new String(bytes));
    }


    public static List<String> getOnlineExecutorIdList(List<OnlineExecutor> onlineExecutorList) {
        List<String> list = new ArrayList<>();
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            list.add(onlineExecutor.getExecutorId());
        }
        return list;
    }

    /**
     * 保留相同的引用而不是复制
     *
     * @param onlineExecutorList
     * @return
     */
    public static List<OnlineExecutor> getOwnTrafficOnlineExecutorList(List<OnlineExecutor> onlineExecutorList) {
        List<OnlineExecutor> list = new ArrayList<>();
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            if (onlineExecutor.isOwnTraffic()) {
                list.add(onlineExecutor);
            }
        }
        return list;
    }

    public static List<String> getOwnTrafficOnlineExecutorIdList(List<OnlineExecutor> onlineExecutorList) {
        List<String> list = new ArrayList<>();
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            if (onlineExecutor.isOwnTraffic()) {
                list.add(onlineExecutor.getExecutorId());
            }
        }
        return list;
    }

    //
    public static List<OnlineExecutor> getDoNotOwnTrafficOnlineExecutorList(List<OnlineExecutor> onlineExecutorList) {
        List<OnlineExecutor> list = new ArrayList<>();
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            if (!onlineExecutor.isOwnTraffic()) {
                list.add(onlineExecutor);
            }
        }
        return list;
    }

    public static List<String> getDoNotOwnTrafficOnlineExecutorIdList(List<OnlineExecutor> onlineExecutorList) {
        List<String> list = new ArrayList<>();
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            if (!onlineExecutor.isOwnTraffic()) {
                list.add(onlineExecutor.getExecutorId());
            }
        }
        return list;
    }

    public static List<OnlineExecutor> copyExecutorList(List<OnlineExecutor> onlineExecutorList) {
        List<OnlineExecutor> $onlineExecutorList = new ArrayList<>();
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            OnlineExecutor $onlineExecutor = new OnlineExecutor();
            $onlineExecutor.setExecutorId(onlineExecutor.getExecutorId());
            $onlineExecutor.setIp(onlineExecutor.getIp());
            $onlineExecutor.setOwnTraffic(onlineExecutor.isOwnTraffic());
            $onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel());
            if (onlineExecutor.getSupportedTaskIdList() != null) {
                $onlineExecutor.setSupportedTaskIdList(new ArrayList<>());
                for (String taskId : onlineExecutor.getSupportedTaskIdList()) {
                    $onlineExecutor.getSupportedTaskIdList().add(taskId);
                }
            }
            if (onlineExecutor.getAssignedTaskItemList() != null) {
                $onlineExecutor.setAssignedTaskItemList(new ArrayList<>());
                for (TaskItem taskItem : onlineExecutor.getAssignedTaskItemList()) {
                    TaskItem $taskItem = new TaskItem();
                    $taskItem.copy(taskItem);
                    $onlineExecutor.getAssignedTaskItemList().add($taskItem);
                }
            }
            $onlineExecutorList.add($onlineExecutor);
        }
        return $onlineExecutorList;
    }
    //##<!-我是优美的功能模块分割线->##


    public static List<TaskItem> getWaitedAssignTaskItemList(//
                                                             CuratorFramework curatorFramework,//
                                                             CronExpressionContextManager cronExpressionContextManager,//
                                                             List<String> enableTaskIdList) throws Exception {//
        List<WaitedAssignTaskInfo> waitedAssignTaskInfoList = new ArrayList<>();
        Map<String, TaskPreAssignTimePoint> taskIdToTaskPreAssignTimePointMap = getTaskIdToTaskPreAssignTimePointMap(curatorFramework);
        /**
         * 获取下次执行的计划时间。 真正进行任务项进行组装不是在这。
         */
        for (String enableTaskId : enableTaskIdList) {
            CronExpressionContext cronExpressionContext = cronExpressionContextManager.taskIdToCronExpressionContextMap.get(enableTaskId);
            if (cronExpressionContext == null) {
                continue;
            } else {
                if (!CronExpressionContext.ENABLE.equals(cronExpressionContext.getEnable())) {
                    continue;
                }
                if (cronExpressionContext.getComputeNextFireTimeMarkDate().getTime() + DateUtils.SECOND * 30 < (new Date()).getTime()) {
                    cronExpressionContext.computeNextFireTime();//recompute
                }
                WaitedAssignTaskInfo waitedAssignTaskInfo = getWaitedAssignTaskInfo(cronExpressionContextManager, enableTaskId, taskIdToTaskPreAssignTimePointMap.get(enableTaskId));
                if (waitedAssignTaskInfo != null) {
                    waitedAssignTaskInfoList.add(waitedAssignTaskInfo);
                }
            }
        }
        List<TaskItem> taskItemList = new ArrayList<>();
        for (WaitedAssignTaskInfo waitedAssignTaskInfo : waitedAssignTaskInfoList) {
            taskItemList.addAll(createTaskItemList(curatorFramework, waitedAssignTaskInfo));
        }
        return taskItemList;
    }

    public static Map<String, TaskPreAssignTimePoint> getTaskIdToTaskPreAssignTimePointMap(CuratorFramework curatorFramework) throws Exception {
        if (curatorFramework.checkExists().forPath(BlockNodePaths.TASK_PRE_ASSIGN_TIME_POINT_NODE_PATH) == null) {
            curatorFramework.create().creatingParentsIfNeeded().forPath(BlockNodePaths.TASK_PRE_ASSIGN_TIME_POINT_NODE_PATH);
            curatorFramework.setData().forPath(BlockNodePaths.TASK_PRE_ASSIGN_TIME_POINT_NODE_PATH, "{}".getBytes(StandardCharsets.UTF_8));
        }
        byte[] bytes = curatorFramework.getData().forPath(BlockNodePaths.TASK_PRE_ASSIGN_TIME_POINT_NODE_PATH);
        Map<String, TaskPreAssignTimePoint> taskIdToTaskPreAssignTimePointMap = (Map<String, TaskPreAssignTimePoint>) JSON.parse(bytes);
        return taskIdToTaskPreAssignTimePointMap;
    }

    /**
     * 判断当前时间点是否可以进行分配的任务
     */
    public static WaitedAssignTaskInfo getWaitedAssignTaskInfo(CronExpressionContextManager cronExpressionContextManager, String enableTaskId, TaskPreAssignTimePoint lastTaskPreAssignTimePoint) {
        CronExpressionContext cronExpressionContext = cronExpressionContextManager.taskIdToCronExpressionContextMap.get(enableTaskId);
        if (cronExpressionContext == null) {
            return null;
        }
        List<TaskPreAssignTimePoint> taskPreAssignTimePointList = cronExpressionContext.getTaskPreAssignTimePointList();
        if (taskPreAssignTimePointList.size() == 0) {
            return null;
        }
        //寻找可以预分配的时间节点，如果已经预分配了那么就不再进行分配了。
        //last预分配节点。
        int lastIndex = taskPreAssignTimePointList.size() - 1;
        Date now = new Date();
        if (lastTaskPreAssignTimePoint == null) {
            for (int i = 0; i <= lastIndex; i++) {//asc
                TaskPreAssignTimePoint taskPreAssignTimePoint = taskPreAssignTimePointList.get(i);
                Date nextFireDate = taskPreAssignTimePoint.getNextFireDate();
                Date nextPreAssignDate = taskPreAssignTimePoint.getNextPreAssignDate();
                if (now.getTime() > nextFireDate.getTime() + 1000L * 5) {
                    continue;
                } else {
                    if (now.getTime() < nextPreAssignDate.getTime()) {
                        return null;
                    } else {
                        cronExpressionContext.setLastPreAssignDate(nextPreAssignDate);
                        cronExpressionContext.setLastRealPreAssignDate(now);
                        //
                        WaitedAssignTaskInfo waitedAssignTaskInfo = new WaitedAssignTaskInfo();
                        waitedAssignTaskInfo.setTaskId(enableTaskId);
                        waitedAssignTaskInfo.setNextFireDate(nextFireDate);
                        waitedAssignTaskInfo.setTaskPreAssignTimePoint(taskPreAssignTimePoint);
                        return waitedAssignTaskInfo;
                    }
                }
            }
            return null;
        } else {
            for (int i = lastIndex; i >= 0; i--) {
                TaskPreAssignTimePoint taskPreAssignTimePoint = taskPreAssignTimePointList.get(i);
                if (taskPreAssignTimePoint.getNextFireDate().getTime() <= lastTaskPreAssignTimePoint.getNextFireDate().getTime()) {
                    continue;//有可能已经预分配了 距离下次分配还有一段时间
                } else {
                    //这代表可以预分配
                    Date nextFireDate = taskPreAssignTimePoint.getNextFireDate();
                    Date nextPreAssignDate = taskPreAssignTimePoint.getNextPreAssignDate();
                    if (now.getTime() > nextFireDate.getTime() + 1000L * 5) {
                        continue;
                    } else {
                        if (now.getTime() < nextPreAssignDate.getTime()) {
                            return null;
                        } else {
                            cronExpressionContext.setLastPreAssignDate(nextPreAssignDate);
                            cronExpressionContext.setLastRealPreAssignDate(now);
                            //
                            WaitedAssignTaskInfo waitedAssignTaskInfo = new WaitedAssignTaskInfo();
                            waitedAssignTaskInfo.setTaskId(enableTaskId);
                            waitedAssignTaskInfo.setNextFireDate(nextFireDate);
                            waitedAssignTaskInfo.setTaskPreAssignTimePoint(taskPreAssignTimePoint);
                            return waitedAssignTaskInfo;
                        }
                    }
                }
            }
            return null;
        }

    }
    //##<!-我是优美的功能模块分割线->##

    /**
     * 创建全新的任务项item list
     */
    public static List<TaskItem> createTaskItemList(CuratorFramework curatorFramework, WaitedAssignTaskInfo waitedAssignTaskInfo) throws Exception {
        List<TaskItem> taskItemList = new ArrayList<>();
        String taskType = waitedAssignTaskInfo.getTaskType();
        //如果是item task or message task那么只有root instance task,基本上就只有root实例。
        if (TaskItem.TASK_TYPE_ITEM_TASK.equals(taskType) || TaskItem.TASK_TYPE_MESSAGE_TASK.equals(taskType)) {
            String taskInstanceId = "root";
            waitedAssignTaskInfo.setTaskInstanceId(taskInstanceId);
            waitedAssignTaskInfo.setTaskRoundId(DateTimeUtil.getTaskRoundId(waitedAssignTaskInfo.getTaskInstanceId()));
            // 新建shardingTotalCount数量的分片
            taskItemList.addAll(createTaskItemList(waitedAssignTaskInfo));
            return taskItemList;
        } else if (TaskItem.TASK_TYPE_BATCH_TASK.equals(taskType)) {
            //batch task允许慢慢处理。
            //如果是batch task 有N+2个任务分片,1任务分配+N执行器+1状态收集(任务分配和状态收集尽可能不要进行随意切换执行器)。
            //如果batch task实例任务分配没有完成 那么只有一个任务分片处理。分配完成后有N+1进行处理。
            //所以针对batch task 需要一种汇报机制。任务分配完成后30秒内进行任务的执行调度，允许不是马上执行。
            String taskInstanceId = UUID.randomUUID().toString();
            waitedAssignTaskInfo.setTaskInstanceId(taskInstanceId);
            waitedAssignTaskInfo.setTaskRoundId(DateTimeUtil.getTaskRoundId(waitedAssignTaskInfo.getTaskInstanceId()));
            //
            BatchTaskInstanceInfo batchTaskInstanceInfo = waitedAssignTaskInfo.getBatchTaskInstanceInfo();
            batchTaskInstanceInfo.setTaskId(waitedAssignTaskInfo.getTaskId());
            batchTaskInstanceInfo.setTaskInstanceId(waitedAssignTaskInfo.getTaskInstanceId());
            batchTaskInstanceInfo.setTaskRoundId(DateTimeUtil.getTaskRoundId(waitedAssignTaskInfo.getTaskInstanceId()));
            //
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskType(taskType);
            taskItem.setTaskId(waitedAssignTaskInfo.getTaskId());
            taskItem.setTaskInstanceId(waitedAssignTaskInfo.getTaskInstanceId());
            taskItem.setTaskRoundId(waitedAssignTaskInfo.getTaskRoundId());
            taskItem.setTaskItemIndex(BatchTaskInstanceInfo.task_item_index_splitting);//进行数据拆分
            taskItem.setLoadLevel(waitedAssignTaskInfo.getLoadLevel());
            taskItem.setNextFireDate(waitedAssignTaskInfo.getNextFireDate());
            taskItem.setTimeoutDateTime(waitedAssignTaskInfo.getTimeoutDateTime());
            taskItem.setLockDateTime(waitedAssignTaskInfo.getLockDateTime());
            taskItemList.add(taskItem);
            batchTaskInstanceInfo.setStage(BatchTaskInstanceInfo.status_data_splitting);
            return taskItemList;
        } else {
            throw new IllegalStateException();
        }

    }

    /**
     * 分片项只能是1-N,不能以0开头
     */
    public static List<TaskItem> createTaskItemList(WaitedAssignTaskInfo waitedAssignTaskInfo) {
        List<TaskItem> taskItemList = new ArrayList<>();
        String taskId = waitedAssignTaskInfo.getTaskId();
        String taskType = waitedAssignTaskInfo.getTaskType();
        String taskInstanceId = waitedAssignTaskInfo.getTaskInstanceId();
        String taskRoundId = waitedAssignTaskInfo.getTaskRoundId();
        int totalTaskItemCount = waitedAssignTaskInfo.getTotalTaskItemCount();
        int loadLevel = waitedAssignTaskInfo.getLoadLevel();
        Date nextFireDate = waitedAssignTaskInfo.getNextFireDate();
        Date timeoutDateTime = waitedAssignTaskInfo.getTimeoutDateTime();
        Date lockDateTime = waitedAssignTaskInfo.getLockDateTime();
        for (int taskItemIndex = 1; taskItemIndex <= totalTaskItemCount; taskItemIndex++) {
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskType(taskType);
            taskItem.setTaskId(taskId);
            taskItem.setTaskInstanceId(taskInstanceId);
            taskItem.setTaskRoundId(taskRoundId);
            taskItem.setTaskItemIndex(taskItemIndex);
            taskItem.setLoadLevel(loadLevel);
            taskItem.setNextFireDate(nextFireDate);
            taskItem.setTimeoutDateTime(timeoutDateTime);
            taskItem.setLockDateTime(lockDateTime);
            taskItemList.add(taskItem);
        }
        return taskItemList;
    }

    /**
     * 正常跑构建
     *
     * @param batchTaskInstanceInfo
     */
    public static void rebuildBatchTaskTaskItem(BatchTaskInstanceInfo batchTaskInstanceInfo) {
        String taskId = batchTaskInstanceInfo.getTaskId();
        String taskType = batchTaskInstanceInfo.getTaskType();
        String taskInstanceId = batchTaskInstanceInfo.getTaskInstanceId();
        String taskRoundId = batchTaskInstanceInfo.getTaskRoundId();
        int totalTaskItemCount = 0;
        int loadLevel = 0;
        Date nextFireDate = new Date();
        Date timeoutDateTime = new Date(nextFireDate.getTime() + DateUtils.HOUR * 6);//超时时间
        Date lockDateTime = timeoutDateTime;//lock时间
        if (BatchTaskInstanceInfo.status_init.equals(batchTaskInstanceInfo.getStage())) {//
            throw new IllegalStateException();
        }
        if (BatchTaskInstanceInfo.status_data_splitting.equals(batchTaskInstanceInfo.getStage())) {//下一步 可能成功 失败 超时
            if (batchTaskInstanceInfo.getTimeoutDateTime().getTime() < (new Date()).getTime()) {
                batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
                batchTaskInstanceInfo.setStage(BatchTaskInstanceInfo.status_data_split_timeout);
                return;
            } else {
                batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(null);//还不能进行重新生成新的执行计划
                return;
            }
        } else if (BatchTaskInstanceInfo.status_data_split_success.equals(batchTaskInstanceInfo.getStage())) {//
            batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(true);
            //
            List<TaskItem> taskItemList = new ArrayList<>();
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskType(taskType);
            taskItem.setTaskId(taskId);
            taskItem.setTaskInstanceId(taskInstanceId);
            taskItem.setTaskRoundId(taskRoundId);
            taskItem.setTaskItemIndex(BatchTaskInstanceInfo.task_item_index_collecting);//任务状态收集
            taskItem.setLoadLevel(loadLevel);
            taskItem.setNextFireDate(nextFireDate);
            taskItem.setTimeoutDateTime(timeoutDateTime);
            taskItem.setLockDateTime(lockDateTime);
            taskItemList.add(taskItem);
            //
            WaitedAssignTaskInfo waitedAssignTaskInfo = new WaitedAssignTaskInfo();
            waitedAssignTaskInfo.setTaskType(taskType);
            waitedAssignTaskInfo.setTaskId(taskId);
            waitedAssignTaskInfo.setTaskInstanceId(taskInstanceId);
            waitedAssignTaskInfo.setTaskRoundId(taskRoundId);
            waitedAssignTaskInfo.setTotalTaskItemCount(totalTaskItemCount);
            waitedAssignTaskInfo.setLoadLevel(loadLevel);
            waitedAssignTaskInfo.setNextFireDate(nextFireDate);
            waitedAssignTaskInfo.setTimeoutDateTime(timeoutDateTime);//超时时间
            waitedAssignTaskInfo.setLockDateTime(lockDateTime);//lock时间

            taskItemList.addAll(createTaskItemList(waitedAssignTaskInfo));
            batchTaskInstanceInfo.setTaskItemList(taskItemList);
            batchTaskInstanceInfo.setStage(BatchTaskInstanceInfo.status_task_execute_and_status_collecting);
            return;
        } else if (BatchTaskInstanceInfo.status_data_split_fail.equals(batchTaskInstanceInfo.getStage())) {//
            batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
            return;
        } else if (BatchTaskInstanceInfo.status_task_execute_and_status_collecting.equals(batchTaskInstanceInfo.getStage())) {
            if (batchTaskInstanceInfo.getTimeoutDateTime().getTime() < (new Date()).getTime()) {
                batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
                batchTaskInstanceInfo.setStage(BatchTaskInstanceInfo.status_task_execute_and_status_collecting_timeout);
                return;
            } else {
                batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(null);//还不能进行重新生成新的执行计划
                return;
            }

        } else if (BatchTaskInstanceInfo.status_task_execute_and_status_collecting_fail.equals(batchTaskInstanceInfo.getStage()) || //
                BatchTaskInstanceInfo.status_task_execute_and_status_collecting_timeout.equals(batchTaskInstanceInfo.getStage())) {//这种情况属于再次清理
            batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);//已经结束了
            return;
        } else if (BatchTaskInstanceInfo.status_task_execute_and_status_collecting_success.equals(batchTaskInstanceInfo.getStage())) {//这种情况属于再次清理
            batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(true);//已经结束了
            List<TaskItem> taskItemList = new ArrayList<>();
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskType(taskType);
            taskItem.setTaskId(taskId);
            taskItem.setTaskInstanceId(taskInstanceId);
            taskItem.setTaskRoundId(taskRoundId);
            taskItem.setTaskItemIndex(BatchTaskInstanceInfo.task_item_index_post_handing);//任务状态收集
            taskItem.setLoadLevel(loadLevel);
            taskItem.setNextFireDate(nextFireDate);
            taskItem.setTimeoutDateTime(timeoutDateTime);
            taskItem.setLockDateTime(lockDateTime);
            taskItemList.add(taskItem);
            return;
        } else if (BatchTaskInstanceInfo.status_data_post_handing.equals(batchTaskInstanceInfo.getStage())) {
            if (batchTaskInstanceInfo.getTimeoutDateTime().getTime() < (new Date()).getTime()) {
                batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
                batchTaskInstanceInfo.setStage(BatchTaskInstanceInfo.status_data_post_hand_timeout);
                return;
            } else {
                batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(null);//还不能进行重新生成新的执行计划
                return;
            }

        } else if (//
                BatchTaskInstanceInfo.status_task_execute_and_status_collecting_success.equals(batchTaskInstanceInfo.getStage()) || //
                        BatchTaskInstanceInfo.status_task_execute_and_status_collecting_fail.equals(batchTaskInstanceInfo.getStage()) || //
                        BatchTaskInstanceInfo.status_task_execute_and_status_collecting_timeout.equals(batchTaskInstanceInfo.getStage())) {//这种情况属于再次清理
            batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);//已经结束了
            return;
        } else {
            //throw new IllegalStateException();
            batchTaskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);//无论那种状态都结束
            return;
        }
    }

    /**
     * 重跑构建
     *
     * @param batchTaskReStartInfo
     * @return
     */
    public static BatchTaskInstanceInfo rebuildBatchTaskTaskItem(BatchTaskReStartInfo batchTaskReStartInfo) {
        String taskType = batchTaskReStartInfo.getTaskType();
        String taskId = batchTaskReStartInfo.getTaskId();
        String taskInstanceId = batchTaskReStartInfo.getTaskInstanceId();
        String taskRoundId = batchTaskReStartInfo.getTaskRoundId();
        int totalTaskItemCount = 0;
        int loadLevel = 0;
        Date nextFireDate = new Date();
        Date timeoutDateTime = new Date(nextFireDate.getTime() + DateUtils.HOUR * 6);//超时时间
        Date lockDateTime = timeoutDateTime;//lock时间
        if (BatchTaskInstanceInfo.status_init.equals(batchTaskReStartInfo.getStage()) ||//
                BatchTaskInstanceInfo.status_data_splitting.equals(batchTaskReStartInfo.getStage()) ||//
                BatchTaskInstanceInfo.status_data_split_fail.equals(batchTaskReStartInfo.getStage()) ||//
                BatchTaskInstanceInfo.status_data_split_timeout.equals(batchTaskReStartInfo.getStage())//
        ) {//
            //如果处于数据拆分中/拆分失败,那么数据拆分程序必须支持重试幂等特性。
            BatchTaskInstanceInfo batchTaskInstanceInfo = new BatchTaskInstanceInfo();
            batchTaskInstanceInfo.setTaskId(taskId);
            batchTaskInstanceInfo.setTaskInstanceId(taskInstanceId);
            batchTaskInstanceInfo.setTaskRoundId(taskRoundId);
            batchTaskInstanceInfo.setRestartAssignTaskItemPlanCreate(true);
            //
            List<TaskItem> taskItemList = new ArrayList<>();
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskType(taskType);
            taskItem.setTaskId(batchTaskReStartInfo.getTaskId());
            taskItem.setTaskInstanceId(taskInstanceId);
            taskItem.setTaskRoundId(taskRoundId);
            taskItem.setTaskItemIndex(BatchTaskInstanceInfo.task_item_index_splitting);//进行数据拆分
            taskItem.setLoadLevel(loadLevel);
            taskItem.setNextFireDate(nextFireDate);
            taskItem.setTimeoutDateTime(timeoutDateTime);
            taskItem.setLockDateTime(lockDateTime);
            taskItemList.add(taskItem);
            batchTaskInstanceInfo.setTaskItemList(taskItemList);
            batchTaskInstanceInfo.setStage(BatchTaskInstanceInfo.status_data_splitting);
            return batchTaskInstanceInfo;
        } else if (BatchTaskInstanceInfo.status_data_split_success.equals(batchTaskReStartInfo.getStage()) ||//
                BatchTaskInstanceInfo.status_task_execute_and_status_collecting.equals(batchTaskReStartInfo.getStage()) ||
                BatchTaskInstanceInfo.status_task_execute_and_status_collecting_fail.equals(batchTaskReStartInfo.getStage()) ||//
                BatchTaskInstanceInfo.status_task_execute_and_status_collecting_timeout.equals(batchTaskReStartInfo.getStage())
        ) {
            BatchTaskInstanceInfo batchTaskInstanceInfo = new BatchTaskInstanceInfo();
            batchTaskInstanceInfo.setTaskId(taskId);
            batchTaskInstanceInfo.setTaskInstanceId(taskInstanceId);
            batchTaskInstanceInfo.setTaskRoundId(taskRoundId);
            batchTaskInstanceInfo.setRestartAssignTaskItemPlanCreate(true);
            //
            List<TaskItem> taskItemList = new ArrayList<>();
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskType(taskType);
            taskItem.setTaskId(taskId);
            taskItem.setTaskInstanceId(taskInstanceId);
            taskItem.setTaskRoundId(taskRoundId);
            taskItem.setTaskItemIndex(BatchTaskInstanceInfo.task_item_index_collecting);//任务状态收集
            taskItem.setLoadLevel(loadLevel);
            taskItem.setNextFireDate(nextFireDate);
            taskItem.setTimeoutDateTime(timeoutDateTime);
            taskItem.setLockDateTime(lockDateTime);
            taskItemList.add(taskItem);
            //
            WaitedAssignTaskInfo waitedAssignTaskInfo = new WaitedAssignTaskInfo();
            waitedAssignTaskInfo.setTaskId(taskId);
            waitedAssignTaskInfo.setTaskType(taskType);
            waitedAssignTaskInfo.setTaskInstanceId(taskInstanceId);
            waitedAssignTaskInfo.setTaskRoundId(taskRoundId);
            waitedAssignTaskInfo.setTotalTaskItemCount(totalTaskItemCount);
            waitedAssignTaskInfo.setLoadLevel(loadLevel);
            waitedAssignTaskInfo.setNextFireDate(nextFireDate);
            waitedAssignTaskInfo.setTimeoutDateTime(timeoutDateTime);//超时时间
            waitedAssignTaskInfo.setLockDateTime(lockDateTime);//lock时间

            taskItemList.addAll(createTaskItemList(waitedAssignTaskInfo));
            batchTaskInstanceInfo.setTaskItemList(taskItemList);
            batchTaskInstanceInfo.setStage(BatchTaskInstanceInfo.status_task_execute_and_status_collecting);
            return batchTaskInstanceInfo;
        } else if (BatchTaskInstanceInfo.status_data_post_handing.equals(batchTaskReStartInfo.getStage()) ||//
                BatchTaskInstanceInfo.status_data_post_hand_fail.equals(batchTaskReStartInfo.getStage()) ||//
                BatchTaskInstanceInfo.status_data_post_hand_timeout.equals(batchTaskReStartInfo.getStage())//
        ) {
            BatchTaskInstanceInfo batchTaskInstanceInfo = new BatchTaskInstanceInfo();
            batchTaskInstanceInfo.setTaskId(taskId);
            batchTaskInstanceInfo.setTaskInstanceId(taskInstanceId);
            batchTaskInstanceInfo.setTaskRoundId(taskRoundId);
            batchTaskInstanceInfo.setRestartAssignTaskItemPlanCreate(true);//无论那种状态都结束

            List<TaskItem> taskItemList = new ArrayList<>();
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskType(taskType);
            taskItem.setTaskId(taskId);
            taskItem.setTaskInstanceId(taskInstanceId);
            taskItem.setTaskRoundId(taskRoundId);
            taskItem.setTaskItemIndex(BatchTaskInstanceInfo.task_item_index_post_handing);//任务状态收集
            taskItem.setLoadLevel(loadLevel);
            taskItem.setNextFireDate(nextFireDate);
            taskItem.setTimeoutDateTime(timeoutDateTime);
            taskItem.setLockDateTime(lockDateTime);
            taskItemList.add(taskItem);
            batchTaskInstanceInfo.setTaskItemList(taskItemList);
            return batchTaskInstanceInfo;
        } else if (BatchTaskInstanceInfo.status_data_post_hand_success.equals(batchTaskReStartInfo.getStage())) {
            BatchTaskInstanceInfo batchTaskInstanceInfo = new BatchTaskInstanceInfo();
            batchTaskInstanceInfo.setTaskId(taskId);
            batchTaskInstanceInfo.setTaskInstanceId(taskInstanceId);
            batchTaskInstanceInfo.setTaskRoundId(taskRoundId);
            batchTaskInstanceInfo.setRestartAssignTaskItemPlanCreate(false);//无论那种状态都结束
            return batchTaskInstanceInfo;
        } else {
            //包括这种状态:BatchTaskInstanceInfo.status_task_execute_and_status_collecting_success.equals(batchTaskReStartInfo.getStage())
            //throw new IllegalStateException();
            BatchTaskInstanceInfo batchTaskInstanceInfo = new BatchTaskInstanceInfo();
            batchTaskInstanceInfo.setTaskId(taskId);
            batchTaskInstanceInfo.setTaskInstanceId(taskInstanceId);
            batchTaskInstanceInfo.setTaskRoundId(taskRoundId);
            batchTaskInstanceInfo.setRestartAssignTaskItemPlanCreate(false);//无论那种状态都结束
            return batchTaskInstanceInfo;
        }
    }

    //

    //##<!-我是优美的功能模块分割线->##

    /**
     * 更新最新的是否拥有流量和支持任务列表
     *
     * @param onlineExecutorList     old image
     * @param lastOnlineExecutorList 最新的
     */
    public static void updateIsOwnTrafficAndSupportedTaskIdList(List<OnlineExecutor> onlineExecutorList, List<OnlineExecutor> lastOnlineExecutorList) {
        for (OnlineExecutor lastOnlineExecutor : lastOnlineExecutorList) {
            for (OnlineExecutor onlineExecutor : onlineExecutorList) {
                if (lastOnlineExecutor.getExecutorId().equals(onlineExecutor.getExecutorId())) {
                    onlineExecutor.setOwnTraffic(lastOnlineExecutor.isOwnTraffic());
                    onlineExecutor.setSupportedTaskIdList(lastOnlineExecutor.getSupportedTaskIdList());
                }
            }
        }
    }

    /**
     * @param onlineExecutorList     old image
     * @param lastOnlineExecutorList 最新的
     */
    public static void addNewOnlineExecutorList(List<OnlineExecutor> onlineExecutorList, List<OnlineExecutor> lastOnlineExecutorList) {
        for (OnlineExecutor lastOnlineExecutor : lastOnlineExecutorList) {
            boolean added = true;//是否需要添加(默认true)
            for (OnlineExecutor onlineExecutor : onlineExecutorList) {
                if (onlineExecutor.getExecutorId().equals(lastOnlineExecutor)) {
                    added = false;
                    break;
                }
            }
            if (added) {
                onlineExecutorList.add(lastOnlineExecutor);
                continue;
            }
        }
    }

    /**
     * 找到并返回oldOnlineExecutorList(newOnlineExecutorList)中 enable/disable task id list,下线executor id/上线没有拥有流量的executor id list/上线拥有流量的executor id list
     */
    public static TaskItemAssignFixPoints getTaskItemAssignFixPoints(TaskItemAssignPresentSituation taskItemAssignPresentSituation) {
        //此处的oldOnlineExecutorList 可能下线 也可能处于上线失去流量也可能上线拥有流量的状态。
        List<OnlineExecutor> oldOnlineExecutorList = taskItemAssignPresentSituation.oldOnlineExecutorList;
        //lastTaskIdList是没有任何意义的，只有lastEnableTaskIdList才进行任务分配
        List<String> lastTaskIdList = taskItemAssignPresentSituation.lastTaskIdList;
        //除开lastEnableTaskIdList都应该下线
        List<String> lastEnableTaskIdList = taskItemAssignPresentSituation.lastEnableTaskIdList;
        //
        List<String> lastOnlineExecutorIdList = taskItemAssignPresentSituation.lastOnlineExecutorIdList;
        List<String> lastOnlineOwnTrafficExecutorIdList = taskItemAssignPresentSituation.lastOnlineOwnTrafficExecutorIdList;
        List<String> lastOnlineDoNotOwnTrafficExecutorIdList = taskItemAssignPresentSituation.lastOnlineDoNotOwnTrafficExecutorIdList;
        //
        //收集oldOnlineExecutorList里面涉及的task id/executor id
        List<String> oldTaskIdList = new ArrayList<>();
        List<String> oldExecutorIdList = new ArrayList<String>();
        Set<String> oldTaskIdSet = new HashSet<>();
        Set<String> oldExecutorIdSet = new HashSet<String>();
        for (OnlineExecutor onlineExecutor : oldOnlineExecutorList) {
            oldTaskIdSet.addAll(onlineExecutor.getSupportedTaskIdList());
            oldExecutorIdSet.add(onlineExecutor.getExecutorId());
        }
        oldTaskIdList.addAll(oldTaskIdSet);
        oldExecutorIdList.addAll(oldExecutorIdSet);
        //排除非交集部分，然后按照新的任务和执行器拓扑结构进行分配任务
        TaskItemAssignFixPoints taskItemAssignFixPoints = new TaskItemAssignFixPoints();
        taskItemAssignFixPoints.setTaskIdList(lastTaskIdList);
        {
            List<String> disableTaskIdList = new ArrayList<String>();
            oldTaskIdList.removeAll(lastEnableTaskIdList);
            disableTaskIdList.addAll(oldTaskIdList);
            taskItemAssignFixPoints.setDisableTaskIdList(disableTaskIdList);
        }
        taskItemAssignFixPoints.setEnableTaskIdList(lastEnableTaskIdList);
        //
        {
            List<String> offlineExecutorIdList = new ArrayList<String>();
            oldExecutorIdList.remove(lastOnlineExecutorIdList);
            offlineExecutorIdList.addAll(oldExecutorIdList);
            taskItemAssignFixPoints.setOfflineExecutorIdList(offlineExecutorIdList);
        }
        //
        taskItemAssignFixPoints.setOnlineExecutorIdList(lastOnlineExecutorIdList);
        taskItemAssignFixPoints.setOnlineOwnTrafficExecutorIdList(lastOnlineOwnTrafficExecutorIdList);
        taskItemAssignFixPoints.setOnlineDoNotOwnTrafficExecutorIdList(lastOnlineDoNotOwnTrafficExecutorIdList);
        //
        return taskItemAssignFixPoints;
    }

    //如果一个任务配置了优先list,那么需要把没有在优先配置列表的先移除,
    //如果没有配置则不管
    public static void fixPreferOnlineExecutorList(List<String> enableTaskIdList, PreferExecutorSettings preferExecutorSettings, List<TaskItem> pickUpEdTaskItemList, List<OnlineExecutor> ownTrafficOnlineExecutorList) {
        for (String taskId : enableTaskIdList) {
            boolean preferExecutorIdListIsConfigured = preferExecutorSettings.taskIdToIsConfiguredPreferExecutorIdListMap.get(taskId); // 是否配置了preferList
            if (preferExecutorIdListIsConfigured) {
                List<String> preferExecutorIdListConfigured = preferExecutorSettings.taskIdToConfiguredPreferExecutorIdListMap.get(taskId); // 配置态的preferList
                if (preferExecutorIdListConfigured != null && preferExecutorIdListConfigured.size() > 0) {
                    for (OnlineExecutor onlineExecutor : ownTrafficOnlineExecutorList) {
                        if (!preferExecutorIdListConfigured.contains(onlineExecutor.getExecutorId())) {
                            List<TaskItem> taskItemList = onlineExecutor.getAssignedTaskItemList();
                            Iterator<TaskItem> iterator = taskItemList.listIterator();
                            while (iterator.hasNext()) {
                                TaskItem taskItem = iterator.next();
                                if (taskItem.getTaskId().equals(taskId)) {
                                    onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                                    iterator.remove();
                                    pickUpEdTaskItemList.add(taskItem);
                                    continue;
                                }
                            }
                        }
                    }
                }
            }
        }

    }

    public static List<OnlineExecutor> getOnlineOwnTrafficButNotAssignedExecutorList(List<OnlineExecutor> newOnlineExecutorList) {
        List<OnlineExecutor> onlineOwnTrafficButNotAssignedExecutorList = newOnlineExecutorList.stream().filter(onlineExecutor -> onlineExecutor.isOwnTraffic() && onlineExecutor.getAssignedTaskItemList().isEmpty()).collect(Collectors.toList());
        return onlineOwnTrafficButNotAssignedExecutorList;
    }

    //##<!-我是优美的功能模块分割线->##
    public static void sortTaskItemList(List<TaskItem> taskItemList) {
        Collections.sort(taskItemList, new Comparator<TaskItem>() {//
            @Override
            public int compare(TaskItem o1, TaskItem o2) {
                int taskIdCompareResult = o1.getTaskId().compareTo(o2.getTaskId());
                if (taskIdCompareResult != 0) {
                    return taskIdCompareResult;
                }
                int taskInstanceIddCompareResult = o1.getTaskInstanceId().compareTo(o2.getTaskInstanceId());
                if (taskInstanceIddCompareResult != 0) {
                    return taskInstanceIddCompareResult;
                }
                int taskRoundIdCompareResult = o1.getTaskRoundId().compareTo(o2.getTaskRoundId());
                if (taskRoundIdCompareResult != 0) {
                    return taskRoundIdCompareResult;
                }
                int loadLevelCompareResult = o1.getLoadLevel() - o2.getLoadLevel();
                return loadLevelCompareResult;
            }
        });
    }
    //##<!-我是优美的功能模块分割线->##

    /**
     * preferExecutorIdList是否配置
     */
    public static boolean preferExecutorIdListIsConfigured(CuratorFramework curatorFramework, String taskId) throws Exception {
        if (curatorFramework.checkExists().forPath(TaskNodePath.getTaskNodePath(taskId)) != null) {
            byte[] bytes = curatorFramework.getData().forPath(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.prefer_executor_id_list));
            if (bytes != null) {
                return new String(bytes, StandardCharsets.UTF_8.name()).trim().length() > 0;
            }
        }
        return false;
    }

    /**
     * 获取当前配置且上线的preferExecutorIdSet
     */
    public static List<String> getPreferExecutorIdListConfigured(CuratorFramework curatorFramework, String taskId, List<String> executorIdList) throws Exception {
        Set<String> preferExecutorIdSet = new HashSet<>();
        if (curatorFramework.checkExists().forPath(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.prefer_executor_id_list)) != null) {
            byte[] bytes = curatorFramework.getData().forPath(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.prefer_executor_id_list));
            if (bytes != null) {
                String[] preferExecutorIds = new String(bytes, StandardCharsets.UTF_8.name()).split(",");
                for (String preferExecutorId : preferExecutorIds) {//preferExecutorIds的数据量相比executorIdList小
                    if (executorIdList.contains(preferExecutorId)) {
                        preferExecutorIdSet.add(preferExecutorId);
                    }
                }
            }
        }
        return new ArrayList<>(preferExecutorIdSet);
    }
    //##<!-我是优美的功能模块分割线->##

    public static void putBackWithPreferExecutorIdListConfigured(List<TaskItem> pickUpEdTaskItemList,//
                                                                 Map<String, List<OnlineExecutor>> taskIdToOwnTrafficOnlineExecutorListMap,//
                                                                 Map<String, Boolean> taskIdToIsConfiguredPreferExecutorIdListMap,//
                                                                 Map<String, List<String>> taskIdToConfiguredPreferExecutorIdListMap) {//
        Iterator<TaskItem> iterator = pickUpEdTaskItemList.iterator();
        while (iterator.hasNext()) {
            TaskItem taskItem = iterator.next();
            String taskId = taskItem.getTaskId();
            // preferList为空不能作为判断是否配置preferList的依据,比如说配置了容器资源,但是全部下线了。
            if (taskIdToIsConfiguredPreferExecutorIdListMap.get(taskId)) {
                List<String> configuredPreferExecutorIdList = taskIdToConfiguredPreferExecutorIdListMap.get(taskId);
                List<OnlineExecutor> preferOwnTrafficOnlineExecutorList = getPreferOwnTrafficOnlineExecutorList(taskIdToOwnTrafficOnlineExecutorListMap, taskId, configuredPreferExecutorIdList);
                // 如果存在preferExecutor，择优放回
                if (preferOwnTrafficOnlineExecutorList != null && !preferOwnTrafficOnlineExecutorList.isEmpty()) {
                    OnlineExecutor onlineExecutor = getOnlineExecutorWithMinTaskTotalLoadLevelOnOnlineExecutor(preferOwnTrafficOnlineExecutorList, taskId);
                    addTaskItemToOnlineExecutor(taskItem, onlineExecutor);
                    iterator.remove();
                } else {
                    // 如果不存在preferExecutor 等到后续进行放回操作，避免不均衡的情况
                }
            }
        }
    }

    public static List<OnlineExecutor> getPreferOwnTrafficOnlineExecutorList(Map<String, List<OnlineExecutor>> taskIdToOwnTrafficOnlineExecutorListMap, String taskId, List<String> configuredPreferExecutorIdList) {
        List<OnlineExecutor> preferOnlineExecutorList = new ArrayList<>();
        List<OnlineExecutor> ownTrafficOnlineExecutorList = taskIdToOwnTrafficOnlineExecutorListMap.get(taskId);
        for (int i = 0; i < ownTrafficOnlineExecutorList.size(); i++) {
            OnlineExecutor onlineExecutor = ownTrafficOnlineExecutorList.get(i);
            if (configuredPreferExecutorIdList.contains(onlineExecutor.getExecutorId())) {
                preferOnlineExecutorList.add(onlineExecutor);
            }
        }
        return preferOnlineExecutorList;
    }

    public static void putBackWithoutPreferExecutorIdListConfigured(List<TaskItem> pickUpEdTaskItemList, Map<String, List<OnlineExecutor>> taskIdToOwnTrafficOnlineExecutorListMap) {
        Iterator<TaskItem> iterator = pickUpEdTaskItemList.iterator();
        while (iterator.hasNext()) {
            TaskItem taskItem = iterator.next();
            List<OnlineExecutor> taskOwnTrafficOnlineExecutorList = taskIdToOwnTrafficOnlineExecutorListMap.get(taskItem.getTaskId());//重点是这段代码 如果没有执行器怎么办???
            if (taskOwnTrafficOnlineExecutorList != null && !taskOwnTrafficOnlineExecutorList.isEmpty()) {
                OnlineExecutor onlineExecutor = getOnlineExecutorWithMinTaskTotalLoadLevelOnOnlineExecutor(taskOwnTrafficOnlineExecutorList, taskItem.getTaskId());
                addTaskItemToOnlineExecutor(taskItem, onlineExecutor);
                iterator.remove();
            }

        }
    }

    /**
     * 获取该任务负荷最小的executor，如果相同，那么取所有任务负荷最小的executor
     */
    public static OnlineExecutor getOnlineExecutorWithMinTaskTotalLoadLevelOnOnlineExecutor(List<OnlineExecutor> taskOwnTrafficOnlineExecutorList, String taskId) {
        OnlineExecutor minTaskTotalLoadLevelOnlineExecutor = null;
        int minTaskTotalLoadLevelOnOnlineExecutor = 0;
        for (int i = 0; i < taskOwnTrafficOnlineExecutorList.size(); i++) {
            OnlineExecutor onlineExecutor = taskOwnTrafficOnlineExecutorList.get(i);
            int taskTotalLoadLevelOnExecutor = getTaskTotalLoadLevelOnExecutor(onlineExecutor, taskId);
            if (minTaskTotalLoadLevelOnlineExecutor == null ||//
                    minTaskTotalLoadLevelOnOnlineExecutor > taskTotalLoadLevelOnExecutor//
                    || minTaskTotalLoadLevelOnOnlineExecutor == taskTotalLoadLevelOnExecutor && minTaskTotalLoadLevelOnlineExecutor.getTotalLoadLevel() > onlineExecutor.getTotalLoadLevel()//
            ) {
                minTaskTotalLoadLevelOnlineExecutor = onlineExecutor;
                minTaskTotalLoadLevelOnOnlineExecutor = taskTotalLoadLevelOnExecutor;
            }
        }
        return minTaskTotalLoadLevelOnlineExecutor;
    }

    /**
     * 计算指定任务在特定executor上的负荷
     */
    public static int getTaskTotalLoadLevelOnExecutor(OnlineExecutor onlineExecutor, String taskId) {
        int taskTotalLoadLevel = 0;
        List<TaskItem> taskItemList = onlineExecutor.getAssignedTaskItemList();
        for (int i = 0; i < taskItemList.size(); i++) {
            TaskItem taskItem = taskItemList.get(i);
            if (taskId.equals(taskItem.getTaskId())) {
                taskTotalLoadLevel += taskItem.getLoadLevel();
            }
        }
        return taskTotalLoadLevel;
    }

    public static void addTaskItemToOnlineExecutor(TaskItem taskItem, OnlineExecutor onlineExecutor) {
        if (onlineExecutor != null) {
            if (isIn(onlineExecutor.getAssignedTaskItemList(), taskItem)) {
                log.error("The shard({}-{}) is running in the executor of {}, cannot be put again", taskItem.getTaskId(), taskItem.getTaskItemIndex(), onlineExecutor.getExecutorId());
            } else {
                onlineExecutor.getAssignedTaskItemList().add(taskItem);
                onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() + taskItem.getLoadLevel());
            }
        } else {
            log.info("No executor to take over the shard: {}-{}", taskItem.getTaskId(), taskItem.getTaskItemIndex());
        }
    }

    public static boolean isIn(List<TaskItem> taskItemList, TaskItem taskItem) {
        for (int i = 0; i < taskItemList.size(); i++) {
            TaskItem $taskItem = taskItemList.get(i);
            if ($taskItem.getTaskId().equals(taskItem.getTaskId()) &&//
                    $taskItem.getTaskInstanceId().equals(taskItem.getTaskInstanceId()) &&//
                    $taskItem.getTaskRoundId().equals(taskItem.getTaskRoundId()) &&//
                    $taskItem.getTaskItemIndex() == taskItem.getTaskItemIndex()) {//
                return true;
            }
        }
        return false;
    }

    //##<!-我是优美的功能模块分割线->##


    /**
     * 刷新onlineExecutorList taskIdList
     */
    public static boolean refreshOnlineExecutorSupportedTaskIdList(CuratorFramework curatorFramework, List<OnlineExecutor> onlineExecutorList, String taskId) throws Exception {
        boolean fixed = false;
        for (int i = 0; i < onlineExecutorList.size(); i++) {
            OnlineExecutor onlineExecutor = onlineExecutorList.get(i);
            if (onlineExecutor.getSupportedTaskIdList() == null) {
                onlineExecutor.setSupportedTaskIdList(new ArrayList<String>());
            }

            String executorTasksNodePath = BlockNodePaths.getExecutorTasksNodePath(onlineExecutor.getExecutorId());
            if (curatorFramework.checkExists().forPath(executorTasksNodePath) != null) {
                onlineExecutor.getSupportedTaskIdList().clear();
                onlineExecutor.getSupportedTaskIdList().addAll(curatorFramework.getChildren().forPath(executorTasksNodePath));
            } else {
                onlineExecutor.getSupportedTaskIdList().clear();
            }
        }
        return fixed;
    }

    /**
     * 移除taskId对应的所有的Task Item List
     */
    public static List<TaskItem> removeTaskItemListOfTask(List<OnlineExecutor> ownTrafficOnlineExecutorList, String taskId) {
        List<TaskItem> removedTaskItemList = new ArrayList<>();
        for (int i = 0; i < ownTrafficOnlineExecutorList.size(); i++) {
            OnlineExecutor onlineExecutor = ownTrafficOnlineExecutorList.get(i);
            Iterator<TaskItem> iterator = onlineExecutor.getAssignedTaskItemList().iterator();
            while (iterator.hasNext()) {
                TaskItem taskItem = iterator.next();
                if (taskId.equals(taskItem.getTaskId())) {//包括该任务下的所有的任务实例
                    onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                    iterator.remove();
                    removedTaskItemList.add(taskItem);
                }
            }
        }
        return removedTaskItemList;
    }


    public static boolean hasTaskItemRunning(String taskId, List<OnlineExecutor> onlineOwnTrafficExecutorList) {
        for (int i = 0; i < onlineOwnTrafficExecutorList.size(); i++) {
            List<TaskItem> taskItemList = onlineOwnTrafficExecutorList.get(i).getAssignedTaskItemList();
            for (int j = 0; j < taskItemList.size(); j++) {
                if (taskItemList.get(j).getTaskId().equals(taskId)) {
                    return true;
                }
            }
        }
        return false;
    }

    public static List<OnlineExecutor> getOnlineExecutorListInPreferOrDisPreferExecutorIdListConfigured(
            String taskId,//
            List<String> preferExecutorIdListConfigured,//
            List<OnlineExecutor> onlineOwnTrafficExecutorList) {//
        List<OnlineExecutor> onlineExecutorList = new ArrayList<>();
        for (OnlineExecutor onlineExecutor : onlineOwnTrafficExecutorList) {
            if (onlineExecutor.getSupportedTaskIdList().contains(taskId) && preferExecutorIdListConfigured.contains(onlineExecutor.getExecutorId())) {
                onlineExecutorList.add(onlineExecutor);
            }
        }
        return onlineExecutorList;
    }

    /**
     * 平衡摘取算法-非多余权重保留模式。保证单个任务均衡不影响所有执行节点。
     *
     * @param onlineExecutorList
     * @param taskItemList
     */
    public static void balancePickUp(List<OnlineExecutor> onlineExecutorList, String taskId, List<TaskItem> taskItemList) {
        // 计算出executor对该任务的平均负荷
        int totalLoadLevel = 0;
        for (TaskItem taskItem : taskItemList) {
            totalLoadLevel += taskItem.getLoadLevel();
        }
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            for (TaskItem taskItem : onlineExecutor.getAssignedTaskItemList()) {
                if (taskId.equals(taskItem.getTaskId())) {
                    totalLoadLevel += taskItem.getLoadLevel();
                }
            }
        }
        int averageTotalLoadLevelAtPerExecutor = totalLoadLevel / onlineExecutorList.size();
        //
        // 摘取executor超出平均负荷的分片
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            int taskTotalLoadLevel = 0;
            Iterator<TaskItem> iterator = onlineExecutor.getAssignedTaskItemList().iterator();
            while (iterator.hasNext()) {
                TaskItem taskItem = iterator.next();
                if (taskId.equals(taskItem.getTaskId())) {
                    //taskTotalLoadLevel + taskItem.getLoadLevel()超过平均权重则不保留。
                    if (averageTotalLoadLevelAtPerExecutor == 0/**很细微的点  可能平均权重为0*/ || taskTotalLoadLevel + taskItem.getLoadLevel() > averageTotalLoadLevelAtPerExecutor) {
                        iterator.remove();
                        onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                        //
                        taskItemList.add(taskItem);
                    } else {
                        //不超过平均权重则保留。
                        taskTotalLoadLevel += taskItem.getLoadLevel();
                    }
                }
            }
        }
    }

    /**
     * getSupportedTaskIdOwnTrafficOnlineExecutorList
     */
    public static List<OnlineExecutor> getSupportedTaskIdOwnTrafficOnlineExecutorList(List<OnlineExecutor> onlineOwnTrafficExecutorList, String taskId) {
        List<OnlineExecutor> list = new ArrayList<>();
        for (OnlineExecutor onlineExecutor : onlineOwnTrafficExecutorList) {
            List<String> taskIdList = onlineExecutor.getSupportedTaskIdList();
            if (onlineExecutor.isOwnTraffic() && taskIdList != null && taskIdList.contains(taskId)) {
                list.add(onlineExecutor);
            }
        }
        return list;
    }


}
